# ******************************************************************#
##author: v_sunao3
##create time: 2023-10-25 10:37:03
# ******************************************************************#


# !/usr/bin/env python3
import sys
import time
from datetime import datetime
from pyhive import hive
import pymysql

# 数据时间通过变量传入
yesterday = '${yyyy-MM-dd-1d}'
print("********************业务日期为：" + str(yesterday))
# 获取当前时间戳
current_timestamp = int(time.time())
# 获取当前时间
current_datetime = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print("********************当前任务运行时间：" + str(current_datetime))
stat_date='${yyyyMMdd-1d}'

hive_database = '${hive_database}'
mysql_database = '${mysql_database}'
hive_table_prefix = '${hive_table_prefix}'

# 先确定该表要创建完成了
result_hive_table = f"{hive_table_prefix}python_mysql_hive_ods_data_cnt_df"
mysql_cnt_hive_table = f"{hive_table_prefix}python_mysql_cnt_df"


# 获取hive连接
def conn_to_hive(hive_database,
                 hive_host='172.25.1.64',
                 hive_port=7001,
                 hive_user='v_sunao3'
                 ):
    return hive.Connection(
        host=hive_host,
        port=hive_port,
        username=hive_user,
        database=hive_database
    )


# 连接mysql数据库
def conn_to_mysql(mysql_database,
                  mysql_host='172.25.1.58',
                  mysql_port=3306,
                  mysql_user='oceanus_user',
                  mysql_password='sqYcvrHc5gpEBH2a',
                  mysql_charset='utf8mb4',
                  ):
    return pymysql.connect(
        host=mysql_host,
        port=mysql_port,
        user=mysql_user,
        password=mysql_password,
        database=mysql_database,
        charset=mysql_charset
    )


# # 连接starrocks数据库
# def conn_to_starrocks(starrocks_host='172.25.1.67',
#                       starrocks_port=9030,
#                       starrocks_user='test',
#                       starrocks_password='srLcHLyUjUrvFf6R',
#                       starrocks_database='test',
#                       starrocks_charset='utf8mb4'
#                       ):
#     return pymysql.connect(
#         host=starrocks_host,
#         port=starrocks_port,
#         user=starrocks_user,
#         password=starrocks_password,
#         database=starrocks_database,
#         charset=starrocks_charset
#     )


# 连接Hive，计算表数据量,传入状态码
def get_mysql_data_count(mysql_table,cursor_hive,stat_date):
    mysql_table_cnt = None
    try:
        # 获取mysql表数据量
        mysql_sql = f"""SELECT mysql_cnt FROM {hive_database}.{mysql_cnt_hive_table} where stat_date='{stat_date}' AND mysql_table_name='{mysql_table}'"""
        print(f"a********************执行计算的sql：{mysql_sql}")
        cursor_hive.execute(mysql_sql)
        mysql_table_cnt = cursor_hive.fetchone()[0]

    except Exception as e:
        print("=" * 100)
        print(f"Failed to connect to Hive: {str(e)}")
        print("=" * 100)
    finally:
        cursor_hive.close()
    return mysql_table_cnt


def get_hive_data_count(hive_table,cursor_hive,stat_date):
    hive_table_cnt = None
    try:
        # 获取Hive表数据量
        hive_sql = f"""SELECT COUNT(stat_date) FROM {hive_database}.{hive_table} where stat_date='{stat_date}'"""
        print(f"********************执行计算的hive_sql：{hive_sql}")
        cursor_hive.execute(hive_sql)
        hive_table_cnt = cursor_hive.fetchone()[0]

    except Exception as e:
        print("+" * 100)
        print(f"Failed to connect to Hive: {str(e)}")
        print("+" * 100)
    finally:
        cursor_hive.close()

    return hive_table_cnt



def check_hive_mysql_data_count(hive_table, mysql_table, conn_hive,threshold = 20):
    '''
    检测数据量是否一致
    :param hive_table: hive表名
    :param mysql_table: mysql表名
    :param conn_hive: hive connect
    :param threshold: 数量相差的阈值
    :return: （打印信息,是否检测 0 未检测 1已检测,[写入sr的数据]）
    '''

    # 计算hive表数据量
    cursor_hive = conn_hive.cursor()

    hive_data_count = get_hive_data_count(hive_table,cursor_hive,stat_date=stat_date)
    mysql_data_count = get_mysql_data_count(mysql_table,cursor_hive,stat_date=stat_date)

    if mysql_data_count is None:
        print(f"=================mysql_data_count:{mysql_data_count}",file=sys.stderr)
        print(f"当天{mysql_table}mysql_data_count is None,快照任务未执行，请检查上游依赖任务,任务失败！！！")
        sys.exit(1)

    # 如果当天还没有同步
    if hive_data_count is None:
        print(f"=================hive_data_count:{hive_data_count}",file=sys.stderr)
        result = [mysql_table, hive_table, stat_date, 0, 'null', mysql_data_count, 'null', "不通过",
                   current_datetime]
    else:
        # 已经同步的数据
        if abs(mysql_data_count - hive_data_count) <= threshold:
            # 设定一个阈值判断，数据量是否一致
            result = [mysql_table, hive_table, stat_date, 1, hive_data_count, mysql_data_count, '是', "通过",
                       current_datetime]
        else:
            result = [mysql_table, hive_table, stat_date, 1, hive_data_count, mysql_data_count, '不', "不通过",
                 current_datetime]

    return result


def get_mysql_tables(cursor_hive):
    mysql_tables = None
    try:
        # 获取Hive表数据量
        hive_sql = f"""SELECT mysql_table_name FROM {hive_database}.{mysql_cnt_hive_table} 
                       where stat_date='{stat_date}' AND mysql_database_name='{mysql_database}'"""
        # print(f"********************执行计算的sql：{sql}")
        cursor_hive.execute(hive_sql)
        mysql_tables = cursor_hive.fetchall()

    except Exception as e:
        print("=" * 100)
        print(f"Failed to connect to Hive: {str(e)}")
        print("=" * 100)
    finally:
        cursor_hive.close()

    return mysql_tables


if __name__ == "__main__":



    # 链接数据库并创建游标
    conn_hive = conn_to_hive(hive_database=hive_database)
    conn_mysql = conn_to_mysql(mysql_database=mysql_database)

    mysql_tables = get_mysql_tables(conn_hive.cursor())
    print(mysql_tables)
    cursor_hive = conn_hive.cursor()
    if mysql_tables is not None:
        try:
            # 删除当前分区数据; alter table ods_sg_zl_python_mysql_hive_ods_data_cnt_df drop partition (stat_date = '20231025')
            # 检测当天任务是否执行
            cursor_hive.execute(f"""select stat_date from {hive_database}.{result_hive_table} where stat_date = '{stat_date}' """)
            if cursor_hive.fetchone() is not None:  # 如果执行了
                # 删除当前分区数据;
                hive_init_sql = f"""alter table {hive_database}.{result_hive_table} drop partition (stat_date = '{stat_date}')"""
                cursor_hive.execute(hive_init_sql)

        except Exception as e:
            print("=" * 100)
            print(f"Failed to init hive partition:{result_hive_table} {str(e)}")
            print("=" * 100)
        finally:
            cursor_hive.close()

        if mysql_tables is not None:
            for mysql_table in list(mysql_tables):
                hive_table = f'{hive_table_prefix}{mysql_table[0]}_df'
                result = check_hive_mysql_data_count(hive_table=hive_table,
                                                     mysql_table=mysql_table[0],
                                                     conn_hive=conn_hive)

                # 插入数据
                cursor_hive = conn_hive.cursor()
                try:
                    """
                    `event_date` string  COMMENT "业务数据日期",
                    `mysql_database_name` string  COMMENT "mysql库名",
                    `mysql_table_name` string  COMMENT "mysql表名",
                    `hive_table_name` string  COMMENT "hive表名",
                    `mysql2hive_status` tinyint  COMMENT "mysql2hive同步状态， 0 未同步 1 已同步",
                    `hive_data_num` bigint   COMMENT "hive表数据量",
                    `mysql_data_num` bigint   COMMENT "msyql表数据量",
                    `is_consistent` string  COMMENT "数据量是否一致 是 或 不",
                    `is_pass` string  COMMENT "是否通过 通过 和 不通过",
                    `script_run_time` string  COMMENT "脚本运行时间"
                    """
                    event_date, mysql_table_name, hive_table_name, mysql2hive_status, hive_field_num, mysql_field_num, field_is_consistent, is_pass, script_run_time = result

                    sql = f"""INSERT INTO table {hive_database}.{result_hive_table} PARTITION(stat_date='{stat_date}') values
                    ("{event_date}","{mysql_database}","{mysql_table_name}","{hive_table_name}",{mysql2hive_status},{hive_field_num},{mysql_field_num},"{field_is_consistent}","{is_pass}","{script_run_time}") """
                    print("1********************", sql)
                    cursor_hive.execute(sql)
                    print('2********************Data inserted successfully.')

                except Exception as e:
                    print('3********************Failed to insert data:', str(e))
                    sys.exit(1)
                finally:
                    # 关闭游标和连接
                    cursor_hive.close()
    else:
        print(f"检测库{mysql_database}失败，请检测上游依赖任务是否完成，后者报错")
        sys.exit(1)

    conn_mysql.close()
    conn_hive.close()
